package org.example.source;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
import org.apache.flink.connector.datagen.source.DataGeneratorSource;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 数据生成器
 */
public class DataGeneratorTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        /**
         * GeneratorFunction
         */
        DataGeneratorSource<String> source = new DataGeneratorSource<>(value -> {
            return "Number:" + value;
        },
                10, // Long类型，自动生成的数字序列（从1自增）的最大值，达到这个值就停止了
                RateLimiterStrategy.perSecond(1), // 限速策略，比如 每秒生成几条数据
                Types.STRING // 返回的类型
        );

        DataStreamSource<String> fengliuklinName = env.fromSource(source, WatermarkStrategy.noWatermarks(), "fengliuklinName");
        fengliuklinName.print();

        env.execute();
    }
}

